//package com.yangzb.sample.hadoop.mapreduce;
//
//import com.yangzb.sample.hadoop.mapreduce.MapUntils.MapBlukLoad;
//import com.yangzb.sample.hadoop.mapreduce.untils.HBaseUntils;
//import com.yangzb.sample.hadoop.mapreduce.untils.PropertiesUtils;
//import org.apache.hadoop.conf.Configuration;
//import org.apache.hadoop.fs.FileSystem;
//import org.apache.hadoop.fs.Path;
//import org.apache.hadoop.hbase.HBaseConfiguration;
//import org.apache.hadoop.hbase.client.*;
//import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
//import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
//import org.apache.hadoop.hbase.mapreduce.ImportTsv;
//import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
//import org.apache.hadoop.mapred.JobConf;
//import org.apache.hadoop.mapreduce.Job;
//import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
//import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
//
//import java.io.File;
//import java.io.FileInputStream;
//import java.io.FileOutputStream;
//import java.io.IOException;
//import java.util.jar.JarEntry;
//import java.util.jar.JarOutputStream;
//import java.util.jar.Manifest;
//
///**
// * Created by yunchen on 2017/3/24.
// * <p>
// * <p>
// * * 思路：
// * 1，rowkey的设计
// * 1）列中的某一个字段作为rwokey
// * 2）列中的某一个字段加上随机数作为rowkey
// * 3）多个列组合加随机数
// * 4）某个列的一部分加随机数
// * 2，列
// * 1）每一列作为一个列
// * 2）多个列组合作为一个列
// * 3，列族
// * 1）通过一个数组或者字符串获取所有数组
// * 2）需要事先定义好哪些列属于哪个列族
// * <p>
// * <p>
// * 工作计划：
// * 1，列族组合功能
// * 2，表创建功能，主要通过列族来创建表，这里涉及到rowkey散列和预分区问题，和rowkey的设计会相关
// * 3，rowkey拼接功能，根据前面列举的几中情况来写相应的函数
// * 4，列的组合功能，这里相对比较复杂，可能需要根据场景来
// */
//
//
//public class BulkLoadMR {
//
//    //TODO 想办法把处理到哪一行的日志打印出来，这样方便排查错误
//    //TODO 还有一些输出信息也打印出，不知道怎么在map中打印出日志来
//    public static void main(String[] args) throws Exception {
//        System.setProperty("HADOOP_USER_NAME", "hdfs");
//
//        //1，获取说需要的变量
//        String config_path = "/Users/yzb/Desktop/source-code/sample/hadoop/mapreduce/src/main/resources/config.properties";
//
//        String tableName = PropertiesUtils.get_NameValues(config_path, "tablename");
//        String srcPath = PropertiesUtils.get_NameValues(config_path, "hdfs_src_path");
//        String hFilePath = PropertiesUtils.get_NameValues(config_path, "hfile_outputdir");
//        //String zkAddress = PropertiesUtils.get_NameValues(config_path,"hbase_zookeeper");
//        //String hbase_master = PropertiesUtils.get_NameValues(config_path,"hbase_master");
//        String hbase_site_path = PropertiesUtils.get_NameValues(config_path, "hbase_site_path");
//        String hdfs_site_path = PropertiesUtils.get_NameValues(config_path, "hdfs_site_path");
//        String columnField = PropertiesUtils.get_NameValues(config_path, "columnField");
//        String columnFamily = PropertiesUtils.get_NameValues(config_path, "columnFamily");
//        String compressType = PropertiesUtils.get_NameValues(config_path, "compressType");
//        Integer regionSplitNum = Integer.valueOf(PropertiesUtils.get_NameValues(config_path, "regionSplitNum"));
//        String splitType = PropertiesUtils.get_NameValues(config_path, "splitType");
//        String rowkeyLine = PropertiesUtils.get_NameValues(config_path, "rowkeyLine");
//        String issubstring = PropertiesUtils.get_NameValues(config_path, "issubstring");
//        String substringbegin = PropertiesUtils.get_NameValues(config_path, "substringbegin");
//        String substringend = PropertiesUtils.get_NameValues(config_path, "substringend");
//        String conbinerowkey = PropertiesUtils.get_NameValues(config_path, "conbinerowkey");
//        String HowNumLine = PropertiesUtils.get_NameValues(config_path, "HowNumLine");
//        String columnsName = PropertiesUtils.get_NameValues(config_path, "columnsName");
//        String columnsName_qua = PropertiesUtils.get_NameValues(config_path, "columnsName_qua");
//
//        //2，创建hbase/hdfs conf
//        Configuration hconf = new Configuration();
//        Configuration conf = HBaseConfiguration.create();
//        conf.addResource(new Path(hbase_site_path));
//        hconf.addResource(new Path(hdfs_site_path));
//
//        //3，创建hbase admin
//        Connection connection = ConnectionFactory.createConnection(conf);
//        Admin admin = connection.getAdmin();
//        FileSystem fs = FileSystem.get(hconf);
//
//        //4，根据提供的表名和列族创建hbase表
//        HBaseUntils hu = new HBaseUntils(admin, connection);
//        String[] cols = columnFamily.split(",");//构建列族数组
//        HBaseUntils.createTableBySplitKeys(tableName, cols, compressType, regionSplitNum, splitType);
//
//        //5，往map函数传递变量，主要有分隔符，rowkey，列
//        conf.set("field", columnField);
//        conf.set("rowkeyLine", rowkeyLine);
//        conf.set("issubstring", issubstring);
//        conf.set("substringbegin", substringbegin);
//        conf.set("substringend", substringend);
//        conf.set("HowNumLine", HowNumLine);
//        conf.set("splitType", splitType);
//        conf.set("columnFamily", columnFamily);
//        conf.set("columnsName", columnsName);
//        conf.set("columnsName_qua", columnsName_qua);
//
//        //6，生成hfile文件
//
//        if (fs.exists(new Path(hFilePath))) {
//            fs.delete(new Path(hFilePath), true);
//        }
//
//        HTable table = new HTable(conf, tableName);
////        table.setWriteBufferSize(6 * 1024 * 1024);
//        Job job = new Job(conf);
//        File jarFile = createTempJar("/Users/yzb/Desktop/source-code/sample/hadoop/mapreduce/target/classes");
//
//        job.setJarByClass(ImportTsv.class);
//        ((JobConf) job.getConfiguration()).setJar(jarFile.toString());
//        job.setMapperClass(MapBlukLoad.Map.class);//指向map class
//        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
//        job.setMapOutputValueClass(Put.class);
//        FileInputFormat.addInputPath(job, new Path(srcPath));
//        //TODO 路径如果存在直接删除，不过要是数据目录删除就麻烦了,可以考虑定义一个没有意义不会使用到的目录
//        FileOutputFormat.setOutputPath(job, new Path(hFilePath));
//        HFileOutputFormat2.configureIncrementalLoad(job, table.getTableDescriptor(), table.getRegionLocator());
//        job.waitForCompletion(true);
//
//        //7，将hfile导入到hbase的表中
//        LoadIncrementalHFiles loadFfiles = new LoadIncrementalHFiles(conf);
//        loadFfiles.doBulkLoad(new Path(hFilePath), admin, table, table.getRegionLocator());
//
//        //8，关闭相关的连接
//        HBaseUntils.close();
//        fs.close();
//
//
//    }
//    public static File createTempJar(String root) throws IOException {
//        if (!new File(root).exists()) {
//            return new File(System.getProperty("java.class.path"));
//        }
//        Manifest manifest = new Manifest();
//        manifest.getMainAttributes().putValue("Manifest-Version", "1.0");
//        final File jarFile = File.createTempFile("test", ".jar", new File(System.getProperty("java.io.tmpdir")));
//        Runtime.getRuntime().addShutdownHook(new Thread() {
//            @Override
//            public void run() {
//                jarFile.delete();
//            }
//        });
//        JarOutputStream out = new JarOutputStream(new FileOutputStream(jarFile), manifest);
//        createTempJarInner(out, new File(root), "");
//        out.flush();
//        out.close();
//        return jarFile;
//    }
//
//    private static void createTempJarInner(JarOutputStream out, File f,
//                                           String base) throws IOException {
//        if (f.isDirectory()) {
//            File[] fl = f.listFiles();
//            if (base.length() > 0) {
//                base = base + "/";
//            }
//            for (int i = 0; i < fl.length; i++) {
//                createTempJarInner(out, fl[i], base + fl[i].getName());
//            }
//        } else {
//            out.putNextEntry(new JarEntry(base));
//            FileInputStream in = new FileInputStream(f);
//            byte[] buffer = new byte[1024];
//            int n = in.read(buffer);
//            while (n != -1) {
//                out.write(buffer, 0, n);
//                n = in.read(buffer);
//            }
//            in.close();
//        }
//    }
//
//}
